package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/9/13 10:10
 * <p>
 * 读取kafka里的ods_base_log：
 * 修复新老访客
 * 分流
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //Configuration configuration = new Configuration();
//        configuration.set(BIND_PORT, "8082");
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        // 从source角度考虑，尽量跟kafka的分区数保持 1：1，效率最高
        env.setParallelism(4);

        //TODO 2.检查点设置
/*        // 2.1 生产环境，间隔是分钟级： 3-10分钟
        env.enableCheckpointing(TimeUnit.SECONDS.toMillis(5), CheckpointingMode.EXACTLY_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        // 2.2 checkpoint超时时间
        // 比如间隔是3分钟，超时可以设为10分钟
        checkpointConfig.setCheckpointTimeout(TimeUnit.SECONDS.toMillis(10));
        // 2.3 checkpoint失败次数
        checkpointConfig.setTolerableCheckpointFailureNumber(3);
        // 2.4 checkpoint最小等待间隔: 从上一次的结束 到 下一次的开始 中间间隔的时间
        // 一般设置为 间隔的 60~80%
        checkpointConfig.setMinPauseBetweenCheckpoints(TimeUnit.SECONDS.toMillis(3));
        // 2.5 设置保留checkpoint
        // checkpoint是给flink程序在重试的时候用的，不是给用户使用的，用户使用的是savepoint
        // 如果想要checkpoint被我们利用起来，设置为 手动取消job后，仍然保存checkpoint
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2.6 Task重试策略
        // 固定延迟重启策略：失败次数 、重试的间隔时间
        // 失败率重启策略: 一段时间内的失败次数 ==》 比率
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, TimeUnit.SECONDS.toMillis(3)));
        // 2.7 指定状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop1:8020/gmall/flink"));
        // 2.8 指定操作HDFS的用户行
        System.setProperty("HADOOP_USER_NAME", "atguigu");*/

        //TODO 3.从Kafka读取数据
        // 3.1 读取kafka
        String sourceTopic = "ods_base_log";
        String groupId = "group_BaseLogApp";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaSourceFunction(sourceTopic, groupId));
        // 3.2 转换数据格式为 JsonObject，方便操作（同时做一下简单的清洗），脏数据也要
        OutputTag<String> dirtyTag = new OutputTag<String>("dirty-data") {
        };
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
                        try {
                            JSONObject jsonObj = JSON.parseObject(value);
                            out.collect(jsonObj);
                        } catch (Exception e) {
                            //将解析失败的 脏数据，放入到侧输出流
                            ctx.output(dirtyTag, value);
                        }

                    }
                }
        );

        // 3.3 打印测试
//        jsonObjDS.print("source>>>>>");

        //TODO 4.修复新老访客
        // 4.1 按照 访客（设备ID） 分组
        SingleOutputStreamOperator<JSONObject> jsonObjWithFixedFlag = jsonObjDS
                .keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<String> firstVisitDateState;
                    private SimpleDateFormat sdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        firstVisitDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstVisitDateState", Types.STRING));
                        sdf = new SimpleDateFormat("yyyyMMdd");
                    }

                    @Override
                    public void processElement(JSONObject value, Context ctx, Collector<JSONObject> out) throws Exception {
                        //1.判断is_new是1还是0
                        String isNew = value.getJSONObject("common").getString("is_new");
                        Long ts = value.getLong("ts");
                        if ("1".equals(isNew)) {
                            // 1.1 如果是1，判断一下状态里是否有值
                            String firstVisitDate = firstVisitDateState.value();
                            if (StringUtils.isNotEmpty(firstVisitDate)) {
                                String currentVisitDate = sdf.format(new Date(ts));
                                // 增加需求：第一次访问，同一天内认为还是新访客
                                if (!currentVisitDate.equals(firstVisitDate)) {
                                    // 如果状态有值，说明状态错了，更新is_new
                                    value.getJSONObject("common").put("is_new", "0");
                                }
                            } else {
                                // 如果状态没有值，说明是新访客，更新状态
                                firstVisitDateState.update(sdf.format(new Date(ts)));
                            }
                        }
                        out.collect(value);
                    }
                });

        // 4.2 测试
//        jsonObjWithFixedFlag.print("is_new fixed>>>>>>");

        //TODO 5.分流: 侧输出流
        // 页面 -主流
        // 启动 -侧流
        // 曝光 -侧流
        // 5.1 分流
        OutputTag<String> startTag = new OutputTag<String>("start") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("display") {
        };
        SingleOutputStreamOperator<String> pageDS = jsonObjWithFixedFlag
                .process(new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject value, Context ctx, Collector<String> out) throws Exception {
                        // 0.先判断是否是 启动日志
                        String startStr = value.getString("start");
                        if (StringUtils.isNotEmpty(startStr)) {
                            // 1. start字段不为空，说明是 启动日志 =》 放到侧输出流
                            ctx.output(startTag, value.toJSONString());
                        } else {
                            // 2. start字段位空，肯定是 页面日志，进一步判断是否含有曝光字段
                            JSONArray displaysArr = value.getJSONArray("displays");
                            // 2.1 将页面日志 放入到 主流
                            out.collect(value.toJSONString());
                            if (displaysArr != null && displaysArr.size() > 0) {
                                // 2.2 说明 曝光字段 存在，属于曝光数据
                                // => 遍历出每条数据
                                // => 补全： page_id、ts 方便后续分析
                                Long ts = value.getLong("ts");
                                String page_id = value.getJSONObject("page").getString("page_id");
                                // 2.3 遍历，并且补全字段，发送到侧输出流
                                for (int i = 0; i < displaysArr.size(); i++) {
                                    JSONObject displayObj = displaysArr.getJSONObject(i);
                                    displayObj.put("ts", ts);
                                    displayObj.put("page_id", page_id);
                                    ctx.output(displayTag, displayObj.toJSONString());
                                }

                            }
                        }
                    }
                });

        // 5.2 测试
        DataStream<String> startDS = pageDS.getSideOutput(startTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displayTag);

        pageDS.print(">>>");
        startDS.print("###");
        displayDS.print("$$$$");


        //TODO 6.输出到dwd的topic
        String pageTopic = "dwd_page_log";
        String startTopic = "dwd_start_log";
        String displayTopic = "dwd_display_log";
        // 6.1 输出 主流 page数据
        pageDS.addSink(MyKafkaUtil.getKafkaSinkFunction(pageTopic));
        // 6.2 输出 侧流 start数据
        startDS.addSink(MyKafkaUtil.getKafkaSinkFunction(startTopic));
        // 6.3 输出 侧流 display数据
        displayDS.addSink(MyKafkaUtil.getKafkaSinkFunction(displayTopic));

        env.execute("dwd_BaseLogApp");
    }
}

/**
 * 需要启动什么？
 * zk、kafka
 * 模拟的jar包、nginx、logger程序
 * BaseLogApp
 * HDFS
 * <p>
 * 执行流程是什么？
 * 模拟数据Mock
 * -> nginx   http://hadoop1:80/applog
 * -> 转发到 logger服务  ，负载均衡
 * -> 写入到 kafka的 ods_base_log
 * -> Flink程序： BaseLogApp 消费、处理
 *      -> 读取 kafka 的 ods_base_log 的数据
 *      -> 修复新老访客
 *      -> 分流: 曝光数据需要判断、炸开、补全字段
 *      -> 输出到dwd层不同的 topic：
 *              dwd_start_log
 *              dwd_page_log
 *              dwd_display_log
 */